package org.apache.flink.runtime.resourceconsumer;

import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.ConsumerID;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.AkkaOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ConsumerManagerOptions;
import org.apache.flink.runtime.slotpool.DeclarativeSlotPoolBridgeServiceFactory;
import org.apache.flink.runtime.slotpool.DeclarativeSlotPoolFactory;
import org.apache.flink.runtime.slotpool.RequestSlotMatchingStrategy;
import org.apache.flink.runtime.slotpool.SimpleRequestSlotMatchingStrategy;
import org.apache.flink.runtime.slotpool.SlotPoolService;
import org.apache.flink.runtime.slotpool.SlotPoolServiceFactory;
import org.apache.flink.util.clock.SystemClock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** Default {@link SlotPoolServiceFactory} implementation. */
/**
 * @类名: DefaultSlotPoolServiceFactory
 * @说明: slot池服务工厂默认实现, 代理模式，实际实现由配置构造
 *
 * @author   leehom
 * @Date	 2023年11月6日 上午9:07:31
 * 修改记录：
 *
 * @see 	 
 * 
 * TODO
 *   > 支持用户自定义slot分配策略
 */
public final class DefaultSlotPoolServiceFactory
        				implements SlotPoolServiceFactory {

    private static final Logger LOG =
            LoggerFactory.getLogger(DefaultSlotPoolServiceFactory.class);

    private final SlotPoolServiceFactory slotPoolServiceFactory;

    private DefaultSlotPoolServiceFactory(
            SlotPoolServiceFactory slotPoolServiceFactory) {
        this.slotPoolServiceFactory = slotPoolServiceFactory;
    }

    @Override
    public SlotPoolService createSlotPoolService(
            ConsumerID jid, DeclarativeSlotPoolFactory declarativeSlotPoolFactory) {
        return slotPoolServiceFactory.createSlotPoolService(jid, declarativeSlotPoolFactory);
    }
    
    // 获取配置，构建slot池服务工厂
    public static SlotPoolServiceFactory fromConfiguration(Configuration configuration) {

    	// 获取配置项
        final Time rpcTimeout =
                Time.fromDuration(configuration.get(AkkaOptions.ASK_TIMEOUT_DURATION));
        final Time slotIdleTimeout =
                Time.milliseconds(configuration.getLong(ConsumerManagerOptions.SLOT_IDLE_TIMEOUT));
        final Time batchSlotTimeout =
                Time.milliseconds(configuration.getLong(ConsumerManagerOptions.SLOT_REQUEST_TIMEOUT));
        // 
        final SlotPoolServiceFactory slotPoolServiceFactory =
                new DeclarativeSlotPoolBridgeServiceFactory(
                        SystemClock.getInstance(),
                        rpcTimeout,
                        slotIdleTimeout,
                        batchSlotTimeout,
                        getRequestSlotMatchingStrategy(configuration));
        return new DefaultSlotPoolServiceFactory(slotPoolServiceFactory);
    }

    // slot匹配策略
    // TODO 支持用户自定义
    @VisibleForTesting
    static RequestSlotMatchingStrategy getRequestSlotMatchingStrategy(
            Configuration configuration) {
    	return SimpleRequestSlotMatchingStrategy.INSTANCE;
    }
    
}
